package com.yyf.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.yyf.kafka.dto.MessageEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * 壹 * @Author: yang
 * 貮 * @Date: 2019/11/5 14:48
 */
@Slf4j
public class ProducerCallback implements ListenableFutureCallback<SendResult<String, MessageEntity>> {


	private final long startTime;
	private final String key;
	private final MessageEntity message;

	public ProducerCallback(long startTime, String key, MessageEntity message) {
		this.startTime = startTime;
		this.key = key;
		this.message = message;
	}

	@Override
	public void onSuccess(SendResult<String, MessageEntity> result) {
		if(ObjectUtils.isEmpty(result)){
			return;
		}

		long endTime = System.currentTimeMillis() - startTime;
		RecordMetadata metadata = result.getRecordMetadata();
		if(!ObjectUtils.isEmpty(metadata)){
			StringBuilder record = new StringBuilder();
			record.append("message(")
					.append("key = ")
					.append(key)
					.append(",")
					.append("message = ")
					.append(JSON.toJSON(message))
					.append(")")
					.append("sent to partition(")
					.append(metadata.partition())
					.append(")")
					.append("with offset(")
					.append(metadata.offset())
					.append("(")
					.append("in ")
					.append(endTime)
					.append(" ms");
			log.info(record.toString());
		}

	}

	@Override
	public void onFailure(Throwable throwable) {
		throwable.printStackTrace();
	}
}
